package com.example.demo.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;

/**
 * @Description
 * @Author liumin0901@sunwayworld.com
 * @Date 2021/5/9 23:18
 */
@Component
public class AvgLatencyConsumerInterceptor implements ConsumerInterceptor<String, String> {

//    private RedisProperties.Jedis jedis;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        long lantency = 0L;
        for (ConsumerRecord record : consumerRecords) {
            lantency += (System.currentTimeMillis() - record.timestamp());
        }
        System.out.println("lantency ="+ lantency);
        redisTemplate.opsForValue().increment("totalLatency", lantency);
        /*jedis.incrBy("totalLatency", lantency);
        long totalLatency = Long.parseLong(jedis.get("totalLatency"));
        long totalSentMsgs = Long.parseLong(jedis.get("totalSentMessage"));*/
        long totalLatency = Long.parseLong(Objects.requireNonNull(redisTemplate.opsForValue().get("totalLatency")).toString());

        long totalSentMsgs = Long.parseLong(Objects.requireNonNull(redisTemplate.opsForValue().get("totalSentMessage")).toString());
        redisTemplate.opsForValue().set("avgLatency", String.valueOf(totalLatency / totalSentMsgs));
//        jedis.set("avgLatency", String.valueOf(totalLatency / totalSentMsgs));
        return consumerRecords;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
